package com.mjf.day4

import com.mjf.day3.{SensorReading, SensorSource}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 1.window操作:
 *  window函数只能在keyBy后面使用
 *  窗口分配器:
 *    .window
 *  窗口函数:基于当前窗口内的数据的，是有界数据集的计算，通常(允许处理延时数据时例外)只在窗口关闭时输出一次
 *    增量聚合函数:来一条数据处理一条数据
 *      ReduceFunction, AggregateFunction
 *    全窗口函数:窗口结束时统一做一次计算
 *      WindowFunction，ProcessWindowFunction
 *
 * 2.window类型:通过窗口分配器来决定
 *  时间窗口:
 *    滚动窗口
 *    滑动窗口:每条数据可以属于多个窗口，属于size/slide个窗口
 *    会话窗口:窗口长度不固定，指定间隔时间没有数据开启新窗口
 *  计数窗口:
 *    滚动窗口
 *    滑动窗口:每条数据可以属于多个窗口，属于size/slide个窗口
 *
 * 3.时间语义:
 *  EventTime:数据产生时间
 *  ProcessingTime(默认):执行操作算子的本地系统时间，与机器相关
 *  IngestionTime:数据进入Flink时间
 *
 * 4.watermark:
 *  定义：事件时间，代表当前时间的进展（之后就不会再来时间戳比watermark里面的数值小的数据了）
 *  作用：主要用来处理乱序数据，一般就是直接定义一个延迟时间，延迟触发窗口操作（延迟指的是当前收到的数据内的时间戳）
 *  延迟时间设置：一般要根据数据的乱序情况来定，通常设置成最大乱序程度（合理取舍效率和数据正确性）
 *  传递：
 *    1）单分区内递增。
 *    2）上游有多个分区时，本分区的watermark为上游所有分区watermark的最小值。
 *
 * 5.窗口:
 *  窗口的两个重要操作：触发计算，清空状态（关闭窗口）
 *  关窗操作：时间进展到窗口关闭时间，事件时间语义下就是watermark达到窗口关闭时间
 *    1）当前Ts最大时间戳-延迟时间 = watermark，watermark大于等于`窗口结束时间`时，关闭窗口
 *    2）如果定义了允许处理迟到数据时，watermark大于等于`窗口结束时间 + 允许延迟时间`时，关闭窗口
 *  注意：watermark只会影响窗口关闭时间，不会影响数据的窗口归属
 *
 * 6.处理乱序数据:
 *  1）设置延迟时间watermark
 *  2）设置窗口允许处理迟到数据的时间（window的allowedLateness方法）
 *  3）将迟到的数据写入侧输出流（window的sideOutputLateData方法）
 *
 */
object WindowExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 设置时间语义，还需要指定数据的哪个字段是EventTime
//    env.getConfig.setAutoWatermarkInterval(100L)  // 设置watermark的生成周期，默认200毫秒

    val inputStream: DataStream[SensorReading] = env.addSource(new SensorSource)

    val resultDataStream: DataStream[SensorReading] = inputStream
      // 数据没有乱序情况时使用`assignAscendingTimestamps`
      //        .assignAscendingTimestamps(_.timestamp * 1000L) // 指定timestamp为EventTime，这里单位是ms
      // 数据存在乱序情况时使用`assignTimestampsAndWatermarks`
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {  // 指定最大乱序(延迟时间)为3s
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L // 提取数据的EventTime
      })
      // 使用自定义的周期性生成watermark的Assigner
      //      .assignTimestampsAndWatermarks(new MyWMAssigner)  // `lateness`也可以定义为参数传入的形式
      // 使用自定义的断点式生成watermark的Assigner
      //      .assignTimestampsAndWatermarks(new MyWMAssigner2)  // `lateness`也可以定义为参数传入的形式
      .keyBy("id")
      //      .window(EventTimeSessionWindows.withGap(Time.minutes(1))) // 会话窗口
      .timeWindow(Time.seconds(15), Time.seconds(5))
      // 允许处理迟到数据。waterMark到达窗口大小时会输出一次计算结果，之后每来一条数据输出一次计算结果
      // waterMark到达`窗口大小 + 允许延迟时间`时关闭窗口
      .allowedLateness(Time.minutes(1))
      .sideOutputLateData(new OutputTag[SensorReading]("late")) // 设置侧输出流保存窗口关闭后来的数据
      .reduce(new ReduceFunction[SensorReading] {
        override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
          SensorReading(value1.id, value1.timestamp.max(value2.timestamp), value1.temperature.min(value2.temperature))
        }
      }) // 增量聚合函数，来一条数据处理一条数据
      //            .apply(new WindowFunction[SensorReading, (String, Long), Tuple, TimeWindow] {
      //              override def apply(key: Tuple, window: TimeWindow, input: Iterable[SensorReading], out: Collector[(String, Long)]): Unit = {
      //                out.collect((window.getStart.toString, input.size)) // 输出窗口起始点和窗口内数据量
      //              }
      //            })  // 全窗口函数，窗口结束时统一做一次计算


    resultDataStream.getSideOutput(new OutputTag[SensorReading]("late"))  // 获取侧输出流，进行后续处理

    resultDataStream.print()

    env.execute("WindowExample")

  }
}

// 自定义一个周期性生成watermark的Assigner
// 系统会周期性的将watermark插入到流中，默认周期是200毫秒，可以使用`env.getConfig.setAutoWatermarkInterval()`方法进行设置
// 参数官方实现的`BoundedOutOfOrdernessTimestampExtractor`
class MyWMAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
  // 需要两个关键参数
  val lateness: Long = 1000L  // 延迟时间
//  var maxTs: Long = Long.MinValue // 当前所有数据中最大的时间戳
  var maxTs: Long = Long.MinValue + lateness  // 加`lateness`防止刚开始没数据的时候`maxTs - lateness`结果为负数

  override def getCurrentWatermark: Watermark = { // 周期性调用
    new Watermark(maxTs - lateness)
  }

  override def extractTimestamp(element: SensorReading, previousElementTimestamp: Long): Long = { // 每来一条数据调用一次
    maxTs = maxTs.max(element.timestamp * 1000L)
    element.timestamp * 1000L // 提取数据的EventTime
  }
}

// 自定义一个断点式生成watermark的Assigner
// 没有时间周期规律，可打断的生成watermark，来一条数据处理一次
class MyWMAssigner2 extends AssignerWithPunctuatedWatermarks[SensorReading] {
  val lateness: Long = 1000L  // 延迟时间
  /**
   * 每来一条数据先调用`extractTimestamp`方法，再调用`checkAndGetNextWatermark`方法
   * @param lastElement 代表本条数据
   * @param extractedTimestamp  提取出来的时间戳
   * @return
   */
  override def checkAndGetNextWatermark(lastElement: SensorReading, extractedTimestamp: Long): Watermark = {  // 每来一条数据调用一次
    if(lastElement.id.equals("sensor_1")) { // 案例：只处理`sensor_1`的watermark
      new Watermark(extractedTimestamp - lateness)
    } else {
      null
    }
  }

  override def extractTimestamp(element: SensorReading, previousElementTimestamp: Long): Long = { // 每来一条数据调用一次
    element.timestamp * 1000L // 提取数据的EventTime
  }
}